package com.corn.flink.lesson3;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * FlinkJdbcSinkOperatorDemo
 *
 * @author JimWu
 * @date 2023/2/28 18:12
 **/
public class FlinkJdbcSinkOperatorDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Person> ds = env.fromElements(
                new Person("alice", 15, "cq"),
                new Person("jack", 18, "cq"),
                new Person("rose", 20, "sh"),
                new Person("mick", 42, "gz"),
                new Person("white", 12, "cq"),
                new Person("maria", 32, "sz"),
                new Person("candy", 56, "bj"),
                new Person("jerry", 28, "wh")
        );
        String sql = "INSERT INTO person VALUES (?,?,?)";
        ds.addSink(JdbcSink.sink(
                sql,
                (statement, data) -> {
                    statement.setString(1, data.name);
                    statement.setInt(2, data.age);
                    statement.setString(3, data.address);
                },
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUrl("jdbc:mysql://localhost:3306/test")
                        .withUsername("root")
                        .withPassword("admin")
                        .build()
        ));

        env.execute();
    }
}
